package rtc.core.net;

import com.coco.base.log.SLog;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * Created by Administrator on 2018/2/9.
 */

public class CoreSocketImpl implements CoreSocket {
    public static final String TAG = "CoreSocketImpl";
    protected final SocketAddress mSocketAddress;
    protected final Transceiver mTransceiver;
    protected final CoreSocket.Callback mCallback;
    protected volatile int mState = STATE_DISCONNECTED;

    private Selector mSelector;
    private SocketChannel mChannel;


    public CoreSocketImpl(String ip,int port,CoreSocket.Callback callback){
        this(new InetSocketAddress(ip,port),callback);
    }

    public CoreSocketImpl(SocketAddress socketAddress,CoreSocket.Callback callback) {
        this(socketAddress,null,callback);
    }

    public CoreSocketImpl(SocketAddress socketAddress,Transceiver transceiver,CoreSocket.Callback callback) {
        this.mSocketAddress = socketAddress;
        this.mTransceiver = (transceiver == null ? new TransceiverImpl(this, (byte) '\n') : transceiver);
        this.mCallback = callback;
    }

    public SocketAddress address(){
        return mSocketAddress;
    }

    public Selector selector(){
        return mSelector;
    }

    public SocketChannel channel(){
        return mChannel;
    }

    @Override
    public boolean connect(long timeout) {
        SLog.i(TAG,"connect mState = %s,timeout = %s",mState,timeout);
        int errorCode = 0;
        if(mState == STATE_DISCONNECTED){
            updateState(STATE_CONNECTING);
            mTransceiver.reset();
            try {
                mSelector = Selector.open();
                mChannel = SocketChannel.open();
                mChannel.configureBlocking(false);
                mChannel.connect(mSocketAddress);
                mChannel.register(mSelector, SelectionKey.OP_CONNECT);
                errorCode = doSelector(timeout);
            } catch (Exception e) {
                e.printStackTrace();
                SLog.e(TAG,"connect Exception",e);
                errorCode = ERROR_UNEXPECTED;
                updateState(STATE_DISCONNECTED);
            }
        }
        final boolean connected = (mState == STATE_CONNECTED);
        if(!connected){
            SLog.w(TAG,"connect failed! errorCode = "+errorCode);
        }
        return connected;
    }

    public void loopSelector(long timeout){
        int count = 0;
        SLog.i(TAG,"loopSelector timeout = "+timeout+",count = "+count);
        while (mState == STATE_CONNECTED){
            try {
                doSelector(timeout);
                mCallback.onSelectorLoop(++count);
            } catch (Exception e) {
                e.printStackTrace();
                SLog.e(TAG,"loopSelector doSelector Exception",e);
            }
        }
        SLog.w(TAG,"loopSelector over");
        close(ERROR_UNEXPECTED);
    }

    public int doSelector(long timeout) throws IOException {
        if (mSelector == null || !mSelector.isOpen()) {
            SLog.d(TAG, "doSelector error, mSelector is null ? " + (mSelector == null));
            return ERROR_UNEXPECTED;
        }
        final int num = mSelector.select(timeout);
        if (num > 0) {
            Set<SelectionKey> selectorKeys = mSelector.selectedKeys();
            Iterator<SelectionKey> it = selectorKeys.iterator();
            while (it.hasNext()) {
                SelectionKey key = it.next();
                it.remove();
                if (!key.isValid()) {
                    continue;
                }
                doKeys(key);
            }
            return ERROR_NONE;
        }else if (mState != STATE_CONNECTED){
            SLog.e(TAG,"doSelector select error,curr not connected,num = " + num);
            close(ERROR_CONNECT_TIME_OUT);
            return ERROR_CONNECT_TIME_OUT;
        }
        return ERROR_NONE;
    }

    private void doKeys(SelectionKey key){
        if(!key.isValid()){
            SLog.w(TAG,"doKeys() key.isValid() is false,cancel it.");
            key.cancel();
            return;
        }
        if(key.isConnectable()){
            SLog.i(TAG,"doKeys() key.isConnectable() is true");
            SocketChannel channel = (SocketChannel) key.channel();
            key.interestOps(SelectionKey.OP_READ);
            try {
                channel.finishConnect();
                updateState(STATE_CONNECTED);
                SLog.i(TAG, "doKeys() SocketChannel finishConnect(), has connect.");
                mCallback.onConnected();
            } catch (Exception e) {
                e.printStackTrace();
                SLog.e(TAG, "doKeys() key.isConnectable(),finishConnect() exception", e);
                key.cancel();
                close(ERROR_CONNECT_FAILED);
            }
        }else if(key.isReadable()){
            SLog.i(TAG,"doKeys() key.isReadable() is true");
            if(mState != STATE_CONNECTED){
                SLog.w(TAG,"doKeys() Readable but state not connected,mState = "+mState);
                return;
            }
            SocketChannel channel = (SocketChannel) key.channel();
            if(channel == null || !channel.isConnected()){
                SLog.e(TAG,"doKeys() Readable but (channel == null || !channel.isConnected())");
                close(ERROR_CONNECT_FAILED);
                return;
            }
            onReadable(channel);
        }else if(key.isWritable()){
            SLog.i(TAG,"doKeys() key.isWritable() is true");
        }else{
            SLog.i(TAG,"doKeys() other state");
        }
    }

    @Override
    public void close() {
        close(ERROR_NONE);
    }

    public void close(int errorCode){
        SLog.i(TAG,"close(),errorCode = %s,mState = %s",errorCode,mState);
        if(mState == STATE_DISCONNECTED){
            return;
        }
        try {
            mChannel.close();
        } catch (Exception e) {
            e.printStackTrace();
            SLog.w(TAG,"mChannel close() exception,errorCode = "+errorCode,e);
        }
        try {
            mSelector.close();
        } catch (Exception e) {
            e.printStackTrace();
            SLog.w(TAG,"mSelector close() exception,errorCode = "+errorCode,e);
        }
        updateState(STATE_DISCONNECTED);
        mCallback.onDisconnected();
    }

    protected void onReadable(SocketChannel channel){
        mTransceiver.onReadable(channel);
    }

    @Override
    public int send(byte[] data, int offset, int length) {
        return mTransceiver.send(data,offset,length);
    }

    @Override
    public int state() {
        return mState;
    }

    private void updateState(int state){
        if(mState != state){
            SLog.i(TAG,"updateState current = %s,new state = %s",mState,state);
            mState = state;
        }
    }

    public static class TransceiverImpl implements CoreSocket.Transceiver{
        private final CoreSocketImpl mSocket;
        private final byte DELIMITER;

        private final ByteBuffer recvBuff = ByteBuffer.allocate(10 * 1024);
        private final RingBuffer recvRing = new RingBuffer(64 * 1024);

        private final ByteBuffer sendBuff = ByteBuffer.allocate(10 * 1024);
        private final RingBuffer sendRing = new RingBuffer(64 * 1024);

        public TransceiverImpl(CoreSocketImpl socket,byte delimiter) {
            this.mSocket = socket;
            this.DELIMITER = delimiter;
        }

        @Override
        public synchronized void reset() {
            recvRing.clear();
            sendRing.clear();
        }

        @Override
        public void onReadable(SocketChannel channel) {
            while (true){
                recvBuff.clear();
                int len = 0;
                try {
                    len = channel.read(recvBuff);
                } catch (Exception e) {
                    e.printStackTrace();
                    SLog.e(TAG,"onReadable channel.read exception",e);
                    mSocket.close(ERROR_CHANNEL_READ);
                }
                if(len == 0){
                    return;
                }else if(len < 0){
                    SLog.e(TAG,"onReadable channel.read len = "+len);
                    mSocket.close(ERROR_CHANNEL_READ);
                    return;
                }else{
                    recvBuff.flip();
                    byte[] buff = recvBuff.array();
                    int beginPos = 0;
                    for(int i = 0 ; i < recvBuff.limit() ; i++){
                        if(buff[i] == DELIMITER){
                            recvRing.write(buff,beginPos,i - beginPos);
                            byte[] packet = new byte[recvRing.readable()];
                            recvRing.read(packet);
                            SLog.i(TAG,"onReadable packet length = "+packet.length);
                            mSocket.mCallback.onData(packet);
                            beginPos = i + 1;
                        }
                        if(i == recvBuff.limit() - 1){
                            recvRing.write(buff,beginPos,recvBuff.limit() - beginPos);
                        }
                    }
                }
            }
        }

        @Override
        public int send(byte[] data, int offset, int length) {
            if(data == null || data.length == 0 || length == 0) return 0;
            if(offset < 0 || length < 0 || (offset + length > data.length)){
                throw new IllegalArgumentException(String.format("params error , offset = %s,length = %s,data.length = %s",offset,length,data.length));
            }
            final SocketChannel channel = mSocket.channel();
            if(channel == null || !channel.isConnected()){
                SLog.e(TAG,"send() but state error ,channel == null || !channel.isConnected()");
                return 0;
            }
            sendRing.write(data,offset,length);
            SLog.i(TAG,"send(),sendRing.write data.length = %s,offset = %s,length = %s",data.length,offset,length);
            if(offset + length == data.length){
                sendRing.write(new byte[]{DELIMITER});
            }
            while (mSocket.mState == STATE_CONNECTED && sendRing.readable() > 0){
                sendBuff.clear();
                final int len = sendRing.read(sendBuff.array());
                if(len <= 0){
                    SLog.e(TAG,"send() error,sendRing.read len = "+ len);
                    continue;
                }
                sendBuff.limit(len);
                try {
                    while (sendBuff.hasRemaining()){
                        channel.write(sendBuff);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    SLog.e(TAG,"send() channel.write Exception",e);
                    mSocket.close(ERROR_CHANNEL_WRITE);
                }
            }
            return length;
        }
    }
}
